package com.alterzhang.userservice.zk;

import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.serialize.SerializableSerializer;
import org.apache.commons.lang.StringUtils;

import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;

public class ZKSynLock {

    private static final String lockPath = "/Lock";

    private static final String zkAddress = "198.181.39.68:2181";

    private ZkClient zkClient = new ZkClient(zkAddress, 100000, 100000, new SerializableSerializer());

    private CountDownLatch cdl;

    private String beforePath;

    private String currentPath;

    public ZKSynLock() {
        if (!this.zkClient.exists(lockPath)) {
            this.zkClient.createPersistent(lockPath);
        }
    }

    public void lock() {
        if (tryLock()) {
            return;
        }
        waitForLock();
        lock();
    }

    private void waitForLock() {

        IZkDataListener iZkDataListener = new IZkDataListener() {
            @Override
            public void handleDataChange(String s, Object o) throws Exception {
                if (null != cdl) {
                    cdl.countDown();
                }
            }

            @Override
            public void handleDataDeleted(String s) throws Exception {

            }
        };
        this.zkClient.subscribeDataChanges(beforePath, iZkDataListener);
        if (this.zkClient.exists(beforePath)) {
            cdl = new CountDownLatch(1);
            try {
                cdl.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        this.zkClient.unsubscribeDataChanges(beforePath, iZkDataListener);
    }

    public boolean tryLock() {
        if (StringUtils.isBlank(currentPath)) {
            currentPath = this.zkClient.createEphemeralSequential(lockPath + "/lock", "lock");
            System.err.println("currentPath====" + currentPath);
        }
        List<String> children = this.zkClient.getChildren(lockPath);
        Collections.sort(children);
        String temStr = children.get(0);
        System.err.println("currentPath="+currentPath+"====temStr="+temStr);
        if (currentPath.equals(lockPath + "/" + temStr)) {
            return true;
        } else {
            int wz = Collections.binarySearch(children, currentPath.substring(6));
            beforePath = lockPath + "/" + children.get(wz - 1);
        }
        return false;
    }

    public void unlock() {
        zkClient.delete(currentPath);
    }

    public static void main(String[] args) {
        ZkClient zkClient = new ZkClient(zkAddress, 10000, 10000, new SerializableSerializer());
        List<String> children = zkClient.getChildren("/Lock");
        Collections.sort(children);
        for (String child : children) {
            System.out.println(child);
        }
        System.out.println(children.get(0));
    }

}
